#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <zlib.h>
#include <errno.h>
#include <pthread.h>

#include <iostream>

#include <log4cxx/logger.h>
#include <log4cxx/xml/domconfigurator.h>

#include "DataSvrClient.h"
#include "DataSvrCommon.h"

#define NUM_OF_THREAD 1

#define WRITE_LEN 1024

using std::cout;
using std::endl;

pthread_mutex_t g_mutex;

string g_log_conf_file;

int parse_args(int argc, char * argv[]) {
        char c;
        while(-1 != (c = getopt(argc, argv, "l:"))) {
                switch(c) {
                        case 'l':
                                g_log_conf_file = optarg;
                                break;

                        default:
                                break;
                }
        }
}

int add_stat(void) {
	static int count = 0;
	static time_t begin_time = time(NULL);
	time_t end_time = time(NULL);
	pthread_mutex_lock(&g_mutex);
	count++;
	pthread_mutex_unlock(&g_mutex);
	if (0 == count % 100) {
		cout << "rate:" << count / (end_time - begin_time + 1) << endl;;
	}
	return 1;
}
	
vector<uint32_t> get_file_crc(const char * file_path) {
	vector<uint32_t> crcs;
        struct stat s;
        int ret = stat(file_path, &s);
        if (0 != ret) {
		cout << "stat file:" << file_path << ", errno:" << errno << endl;
		return crcs; 
	}
        char * buf = new char[s.st_size];
        FILE * fp = fopen(file_path, "r");
        fread(buf, s.st_size, 1, fp);
        fclose(fp);
        for (int i = 0; i < s.st_size; i+= 1024 * 1024) {
                int len = 1024 * 1024;
                int left = s.st_size - i;
                if (left < 1024 * 1024) len = left;
                uLong crc_init = crc32(0L, NULL, 0);
                uint32_t crc = crc32(crc_init, (Bytef *)&buf[i], len);
		crcs.push_back(crc);
        }
        delete[] buf;
	return crcs;
}

void dump_crcs(vector<uint32_t> crcs) {
	for (vector<uint32_t>::const_iterator iter = crcs.begin(); crcs.end() != iter; iter++) {
		cout << *iter << endl;
	}
}

bool is_crcs_equal(vector<uint32_t> crcs1, vector<uint32_t> crcs2) {
	if (crcs1.size() != crcs2.size()) {
		cout << "vector size not equal, csc1 size:" << crcs1.size() << ", crcs2 size:" << crcs2.size() << endl;
		dump_crcs(crcs1);
		cout << "------------------------" << endl;
		dump_crcs(crcs2);
		return false;
	}
	int i = 0;
	for (vector<uint32_t>::const_iterator iter = crcs1.begin(); crcs1.end() != iter; iter++) {
		if (*iter != crcs2[i]) {
			cout << "vector value not equal" << endl;
			dump_crcs(crcs1);
			cout << "------------------------" << endl;
			dump_crcs(crcs2);
			return false;
		}
		i++;
	}
	return true;
}

void dump_buf(const char * buf, int len) {
	if (NULL == buf) return; 
	for (int i = 0; i < len; i++) {
		cout << buf[i];
	}
	cout << endl;
}


int test_replicate() {
	CnfSvrClient csc;
	int ret = csc.init("127.0.0.1", 3501,  100,  5, 2, 2, -1);
	if (ret < 0) {
		cout << "cnf svr client init errror, ret:" << ret << endl;
		return -1;
	}
	DataSvrClientGroup dscg;
	dscg.init(&csc);
	DataSvrClient * dsc1 = dscg.get_data_svr_client(1);
	DataSvrClient * dsc2 = dscg.get_data_svr_client(2);
	if (NULL == dsc1 || NULL == dsc2) {
		cout << "could not get ds client" << endl;
		return -2;
	}
	ret = dsc1->create_file(1, 1);
	if (ret < 0) {
		cout << "ds1 create file err, ret:" << ret << endl;
//		return -3;
	}
	int buf_size = 3 * 1024;
	char * buf = new char[buf_size];
	for(int i = 0; i < buf_size; i++) buf[i] = 'a';
	int begin_pos = 0;
	ret = dsc1->write_file(1,1, begin_pos, buf_size, buf, DS_WRITE_MODE_NORMAL);
	if (ret < 0) {
		cout << "ds1 write file err, ret:" << ret << endl;
		return -4;
	}
	DS_FILE_CONTROL_INFO info1, info2;
	int64_t file_size1, file_size2;
	ret = dsc1->get_file_control_info(1, 1, info1, file_size1);
	if (ret < 0) {
		cout << "ds1 get file control info ret:" << ret << endl;
		return -5;
	}
	ret = dsc2->replicate(1, 1, 1, file_size1, info1.crcs);
	if (ret < 0) {
		cout << "ds2 replicate error, ret:" << ret << endl;
		return -5;
	}
	sleep(5);
	ret = dsc2->get_file_control_info(1, 1, info2, file_size2);
	if (ret < 0) {
		cout << "ds2 get file control info ret:" << ret << endl;
		return -6;
	}
	if (!is_crcs_equal(info1.crcs, info2.crcs)) {
		cout << "replicate error!" << endl;
		return -7;
	}
	return 1;
}

int test_basic() {
	DataSvrClient c;
	c.init("127.0.0.1", 3502, 10, 5, 2, 3, -1, 1);
	while(true) {
		int dir_id = rand();
		int file_id = rand();
		int ret = c.create_file(dir_id, file_id);
		if (ret < 0) cout << "create file ret:" << ret << endl;
		char * buf = new char[WRITE_LEN];
		for (int i =0; i < WRITE_LEN; i++) {
			buf[i] = 'a' + i % 26;
		}
		int write_pos = rand() / 81920100;
		//ret = c.write_file(1, 1, write_pos, WRITE_LEN, buf, DS_WRITE_MODE_APPEND);
		ret = c.write_file(dir_id, file_id, 0, WRITE_LEN, buf, DS_WRITE_MODE_APPEND);
		if (ret < 0) cout << "write file ret:" << ret << endl;
		DS_FILE_CONTROL_INFO info;
		int64_t file_size;
		ret = c.get_file_control_info(dir_id, file_id, info, file_size);
		if (ret < 0) cout << "get file control info ret:" << ret << endl;
		char * read_buf = new char[WRITE_LEN];
		ret = c.read_file(dir_id, file_id, 0, WRITE_LEN, read_buf);
		if (ret < 0) cout << "read file ret:" << ret << endl;
		if (ret > 0) {
			for (int i = 0; i < WRITE_LEN; i++) {
				if (buf[i] != read_buf[i]) {
					cout << "...............oops write and read content not equal" << endl;
					dump_buf(buf, WRITE_LEN);
					cout << "------------------------------------------------------------------------" << endl;
					dump_buf(read_buf, WRITE_LEN);
					break;
				}
			}
		}
		ret = c.delete_file(dir_id, file_id);
		if (ret < 0) cout << "delete file ret:" << ret << endl;
		delete[] buf;
		delete[] read_buf;
		add_stat();
		return 1;
		/*
		vector<uint32_t> server_crcs = info.crcs;
		vector<uint32_t> client_crcs;
		client_crcs = get_file_crc("/data1/xlk/projects/sfs/data_svr/server/src/ds_data/1/0/1");
		if (!is_crcs_equal(client_crcs, server_crcs)) {
			cout << "...........oops" << endl;
			return NULL;
		}
		*/
	}

}

void * test_thread_routine(void * arg) {
	test_replicate();
	//test_basic();
}

int start_test() {
	srand(time(NULL));	
	pthread_t t[NUM_OF_THREAD];
	for (int i = 0; i < NUM_OF_THREAD; i++) {
		pthread_create(&t[i], NULL, test_thread_routine, NULL);
	}
	for (int i = 0; i < NUM_OF_THREAD; i++) {
		pthread_join(t[i], NULL);
	}
	return 1;	
}

int init() {
	log4cxx::xml::DOMConfigurator::configure(g_log_conf_file.c_str());
	pthread_mutex_init(&g_mutex, NULL);
	return 1;
}

int main(int argc, char * argv[]) {
	parse_args(argc, argv);
	init();
	start_test();
}
